package com.shujia.opt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.util.Random


object Code03SparkFilterJoin {
  def main(args: Array[String]): Unit = {
    /**
     * 数据倾斜：
     * 当数据做关联时，如果其中两张表的数据都比较大，并且其中一张表的数据Key分布不均
     *
     * 解决方案：
     * 1.需要将分布不均的数据采集出来
     * 2.对分布不均的Key数据进行添加随机数
     * 3.对分布不均的Key数据进行关联
     * 4.其他均匀数据正常关联即可
     *
     */

    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("join"))


    // 该RDD中存在部分的Key分布不均
    val joinRDD1: RDD[(String, String)] = sc.parallelize(List(("hello", "11"), ("hello", "11"), ("hello", "11"), ("hello", "11"), ("hello", "11"), ("hello", "11"), ("hello", "11"), ("hello", "11"),
      ("world", "12"), ("spark", "13"), ("hadoop", "14")
    ))

    val joinRDD2: RDD[(String, String)] = sc.parallelize(List(("hello", "1"),
      ("world", "2"), ("spark", "3"), ("hadoop", "4")
    ))


    // 将可能发生数据倾斜的Key过滤出来
    val topMap = joinRDD1
      //      .sample(true, 0.8)
      .groupBy(_._1)
      .mapValues(_.size)
      .map(x => (x._2, x._1))
      .top(1)
      .map {
        case (value, key) => (key, value)
      }.toMap
    println(topMap)

    val mapRandomRes: RDD[(String, String)] = joinRDD1
      .filter {
        // 如果过滤出来是分布不均的Key，那么单独处理
        case (key, value) => {
          topMap.contains(key)
        }
      }
      .map {
        // 对分布不均匀的Key添加随机数进行打散操作
        case (key, value) => {
          (s"${new Random().nextInt(4)}_${key}", value)
        }
      }

    val joinRandomRes: RDD[(String, (String, String))] = joinRDD2
      .filter {
        case (key, value) => {
          topMap.contains(key)
        }
      }
      .flatMap {
        case (key, value) => {
          List(
            (s"0_$key", value),
            (s"1_$key", value),
            (s"2_$key", value),
            (s"3_$key", value)
          )
        }
      }.join(
      mapRandomRes
    ).map {
      case (key, value) => {
        (key.substring(2), value)
      }
    }


    joinRDD1
      .filter {
        // 如果过滤出来是分布均匀的Key，正常处理
        case (key, value) => {
          !topMap.contains(key)
        }
      }
      .join(joinRDD2)
      // 将关联好的结果进行join操作
      .union(joinRandomRes)
      .foreach(println)


  }
}
